use std::mem;

use libp2p::{
    Swarm,
    gossipsub::{self, IdentTopic},
    identify,
    kad::{self, GetProvidersOk, QueryId, QueryResult},
    mdns,
    multiaddr::Protocol,
    relay,
    request_response::{self},
};
use tokio::fs;
use tracing::{debug, error, info, warn};

use crate::app::{
    FileChunkData, FileProcessResult, GOSSIPSUB_PUBLIC_FILES_TOPIC_NAME, LOG_TARGET,
    P2pNetworkBehavior, P2pService,
    p2p::{
        FileChunkDownloadRequest, FileChunkDownloadResponse, MetadataDownloadRequest,
        MetadataDownloadResponse,
    },
    store::{self, PublicPublishedFile, PublishedFileRecord},
};

impl<FileStore: store::Store> P2pService<FileStore> {
    /// 处理 libp2p 中 identify 协议产生的事件。
    /// identify 协议允许节点交换自身的身份信息，如公钥、支持的协议、监听地址等
    pub(super) fn process_identify_event(
        &self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        event: identify::Event,
    ) {
        match event {
            identify::Event::Received { info, peer_id, .. } => {
                debug!(target: LOG_TARGET, "Received identify info: {:?}", info);
                // 检查接收到的节点支持的协议列表中是否包含中继协议
                // relay::HOP_PROTOCOL_NAME 是中继协议的名称
                // 如果存在，则认为该节点是中继（relay）节点
                let is_relay = info
                    .protocols
                    .iter()
                    .any(|p| p == &relay::HOP_PROTOCOL_NAME);

                // 将该节点添加到 gossipsub 协议的显式对等节点列表中
                // 这样后续可以通过 gossipsub 协议向该节点发送消息
                swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer_id);

                for addr in info.listen_addrs {
                    // 将该节点的地址添加到 Kademlia DHT 协议的路由表中
                    // 方便后续通过 DHT 查找该节点
                    swarm
                        .behaviour_mut()
                        .kademlia
                        .add_address(&peer_id, addr.clone());

                    // 如果该节点是中继节点
                    if is_relay {
                        // 在地址后面添加 P2P 协议部分，包含该节点的 PeerID
                        // 若添加失败，将错误转换为 ServerError 并返回
                        let addr = match addr.with_p2p(peer_id) {
                            Ok(addr) => addr,
                            Err(e) => {
                                debug!(
                                    target: LOG_TARGET,
                                    "Failed to add P2P protocol to address {e:?}",
                                );
                                continue;
                            }
                        };
                        let listen_addr = addr
                            // 再添加 P2P 中继电路协议部分
                            .with(Protocol::P2pCircuit);

                        info!(target: LOG_TARGET, "Trying to listen to relay with address {:?}", listen_addr);
                        // 尝试在该中继地址上进行监听
                        // 若监听失败，将错误转换为 ServerError 并返回
                        if let Err(e) = swarm.listen_on(listen_addr.clone()) {
                            warn!(
                                target: LOG_TARGET,
                                "Failed to listen on relay ({:?}): {:?}",
                                listen_addr,
                                e
                            )
                        }
                    }
                }
            }
            identify::Event::Sent { .. } => {
                debug!(target: LOG_TARGET, "Sent identify info");
            }
            _ => {
                debug!(target: LOG_TARGET, "identify event {:?}",  event);
            }
        }
    }

    /// 处理 mdns 协议发现的事件
    pub(super) fn process_mdns_event(
        &self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        event: mdns::Event,
    ) {
        match event {
            mdns::Event::Discovered(new_peers) => {
                for (peer_id, addr) in new_peers {
                    info!(target: LOG_TARGET, "[mDNS] Discovered {peer_id} at {addr}");
                    swarm.add_peer_address(peer_id, addr.clone());
                    swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
                    swarm.behaviour_mut().gossipsub.add_explicit_peer(&peer_id);
                }
            }
            mdns::Event::Expired(items) => {
                debug!(target: LOG_TARGET, "expired mdns items: {:?}", items);
            }
        }
    }

    /// 处理公共文件发布事件
    /// 当有新的公共文件发布时，将该文件发布到 gossipsub 网络中，其他peer就能收到广播，然后去下载？
    pub fn handle_public_file(
        &self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        file: &PublishedFileRecord,
    ) {
        let data: PublicPublishedFile = file.into();
        let data: Vec<u8> = match data.try_into() {
            Ok(data) => data,
            Err(err) => {
                error!(target: LOG_TARGET, "convert published file record to vec error: {:?}", err);
                return;
            }
        };
        if let Err(err) = swarm
            .behaviour_mut()
            .gossipsub
            .publish(IdentTopic::new(GOSSIPSUB_PUBLIC_FILES_TOPIC_NAME), data)
        {
            error!(target: LOG_TARGET, "gossipsub publish public file error: {:?}", err);
        }
    }

    /// 处理 gossipsub 事件
    pub(super) fn process_gossipsub_event(
        &mut self,
        _swarm: &mut Swarm<P2pNetworkBehavior>,
        event: gossipsub::Event,
    ) {
        match event {
            gossipsub::Event::Message {
                propagation_source: _propagation_source,
                message_id: _message_id,
                message,
            } => {
                // 应为自己也订阅了该消息，如果是自己发布的，就过滤掉自己
                info!(target: LOG_TARGET, "[gossipsub] Received message: {:?}", message);
                if message.topic.as_str() == GOSSIPSUB_PUBLIC_FILES_TOPIC_NAME {
                    let data: PublicPublishedFile = match message.data.try_into() {
                        Ok(data) => data,
                        Err(err) => {
                            error!(target: LOG_TARGET, "convert gossipsub message data to public published file error: {:?}", err);
                            return;
                        }
                    };
                    info!(target: LOG_TARGET, "[gossipsub] Received public file: {:?}", &data);
                    // 内存中记录已经发布了的文件信息
                    self.public_files.insert(data);
                }
            }
            _ => {
                debug!(target: LOG_TARGET, "[gossipsub] event {:?}",  event);
            }
        }
    }

    /// 处理文件下载事件
    pub(super) async fn process_file_download_event(
        &mut self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        event: request_response::Event<FileChunkDownloadRequest, FileChunkDownloadResponse>,
    ) {
        match event {
            request_response::Event::Message {
                peer,
                connection_id,
                message,
            } => {
                info!(target: LOG_TARGET, "Received file download request from peer: {:?}, connection_id: {:?}, message: {:?}", peer, connection_id, message);
                match message {
                    request_response::Message::Request {
                        request_id,
                        request,
                        channel,
                    } => {
                        info!(target: LOG_TARGET, "Received file download request from peer: {:?}, connection_id: {:?}, request_id: {:?}, request: {:?}", peer, connection_id, request_id, request);

                        // we already have the whole file, so we have all the chunks too
                        // 节点分3种情况讨论
                        // 1.当前节点有整个文件，也就有整个文件的chunk
                        // 2.当前节点没有有整个文件，只有整个文件的部分chunk
                        // 3.完全没有文件的chunk，节点不处理

                        // 1.当前节点有整个文件，也就有整个文件的chunk
                        if let Ok(true) = self.file_store.published_file_exists(request.file_id) {
                            info!(target: LOG_TARGET, "File {:?} exists", request.file_id);
                            match self
                                .file_store
                                .fetch_published_file_chunk_path(request.file_id, request.chunk_id)
                            {
                                Ok(Some(chunk_path)) => {
                                    match fs::read(chunk_path).await {
                                        Ok(chunk_data) => {
                                            info!(target: LOG_TARGET, "Read file chunk: {:?} successfully", chunk_data);
                                            if let Err(e) =
                                                swarm.behaviour_mut().file_download.send_response(
                                                    channel,
                                                    FileChunkDownloadResponse::Success(
                                                        FileChunkData {
                                                            file_id: request.file_id,
                                                            chunk_id: request.chunk_id,
                                                            data: chunk_data,
                                                        },
                                                    ),
                                                )
                                            {
                                                error!(target: LOG_TARGET, "Failed to send file chunk response: {:?}", e);
                                            }
                                        }
                                        Err(e) => {
                                            error!(target: LOG_TARGET, "Failed to read file chunk: {:?}", e);
                                            if let Err(e) =
                                                swarm.behaviour_mut().file_download.send_response(
                                                    channel,
                                                    FileChunkDownloadResponse::Error(format!(
                                                        "Failed to read file chunk: {:?}",
                                                        e
                                                    )),
                                                )
                                            {
                                                error!(target: LOG_TARGET, "Failed to send file chunk response: {:?}", e);
                                            }
                                        }
                                    };
                                }
                                Ok(None) => {
                                    error!(target: LOG_TARGET, "File {:?} chunk {:?} path not found", request.file_id, request.chunk_id);
                                    if let Err(e) =
                                        swarm.behaviour_mut().file_download.send_response(
                                            channel,
                                            FileChunkDownloadResponse::Error(format!(
                                                "File {:?} chunk {:?} path not found",
                                                request.file_id, request.chunk_id
                                            )),
                                        )
                                    {
                                        error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                    }
                                }
                                Err(e) => {
                                    if let Err(e) =
                                        swarm.behaviour_mut().file_download.send_response(
                                            channel,
                                            FileChunkDownloadResponse::Error(format!(
                                                "Failed to fetch file chunk path: {:?}",
                                                e
                                            )),
                                        )
                                    {
                                        error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                    }
                                }
                            }
                            return;
                        }

                        // we have our chunk downloaded as a pending download

                        // 2.当前节点没有有整个文件，只有整个文件的部分chunk
                        if let Ok(true) =
                            self.file_store.check_downloaded_chunks_in_pending_download(
                                request.file_id,
                                request.chunk_id,
                            )
                        {
                            match self.file_store.fetch_pending_downloaded_chunk_path(
                                request.file_id,
                                request.chunk_id,
                            ) {
                                Ok(Some(chunk_path)) => {
                                    match fs::read(chunk_path).await {
                                        Ok(chunk_data) => {
                                            info!(target: LOG_TARGET, "Read file chunk: {:?} successfully", chunk_data);
                                            if let Err(e) =
                                                swarm.behaviour_mut().file_download.send_response(
                                                    channel,
                                                    FileChunkDownloadResponse::Success(
                                                        FileChunkData {
                                                            file_id: request.file_id,
                                                            chunk_id: request.chunk_id,
                                                            data: chunk_data,
                                                        },
                                                    ),
                                                )
                                            {
                                                error!(target: LOG_TARGET, "Failed to send file chunk response: {:?}", e);
                                            }
                                        }
                                        Err(e) => {
                                            error!(target: LOG_TARGET, "Failed to read file chunk: {:?}", e);
                                            if let Err(e) =
                                                swarm.behaviour_mut().file_download.send_response(
                                                    channel,
                                                    FileChunkDownloadResponse::Error(format!(
                                                        "Failed to read file chunk: {:?}",
                                                        e
                                                    )),
                                                )
                                            {
                                                error!(target: LOG_TARGET, "Failed to send file chunk response: {:?}", e);
                                            }
                                        }
                                    };
                                }
                                Ok(None) => {
                                    error!(target: LOG_TARGET, "File {:?} chunk {:?} path not found", request.file_id, request.chunk_id);
                                    if let Err(e) =
                                        swarm.behaviour_mut().file_download.send_response(
                                            channel,
                                            FileChunkDownloadResponse::Error(format!(
                                                "File {:?} chunk {:?} path not found",
                                                request.file_id, request.chunk_id
                                            )),
                                        )
                                    {
                                        error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                    }
                                }
                                Err(e) => {
                                    if let Err(e) =
                                        swarm.behaviour_mut().file_download.send_response(
                                            channel,
                                            FileChunkDownloadResponse::Error(format!(
                                                "Failed to fetch file chunk path: {:?}",
                                                e
                                            )),
                                        )
                                    {
                                        error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                    }
                                }
                            }
                        }

                        // 3.完全没有文件的chunk, 节点不处理
                    }
                    request_response::Message::Response {
                        request_id,
                        response,
                    } => {
                        info!(target: LOG_TARGET, "Received file download response from peer: {:?}, connection_id: {:?}, request_id: {:?}, response: {:?}", peer, connection_id, request_id, response);

                        match response {
                            FileChunkDownloadResponse::Success(chunk_data) => {
                                info!(target: LOG_TARGET, "Received file chunk: {:?}", chunk_data);

                                if let Some(file_chunk_request_data) = self
                                    .find_file_chunk_download_request_with_request_id_mut(
                                        request_id,
                                    )
                                {
                                    info!(target: LOG_TARGET, "Send file chunk download response, request_id: {:?}, file chunk download success, sender: {:?}", request_id, file_chunk_request_data.result);
                                    // 这里使用mem::take取result的值，如果result有值，则该值被取出，result赋值为None
                                    // 或者使用 metadata_request_data.take() 也可以，底层是调用了 mem::replace()
                                    if let Some(sender) =
                                        mem::take(&mut file_chunk_request_data.result)
                                    {
                                        if let Err(e) = sender.send(chunk_data.data) {
                                            info!(target: LOG_TARGET, "Failed to send file chunk id to request result channel: {:?}", e);
                                            return;
                                        }
                                    }
                                }
                            }
                            FileChunkDownloadResponse::Error(err) => {
                                error!(target: LOG_TARGET, "Failed to download file chunk: {:?}", err);
                            }
                        }

                        self.remove_file_chunk_download_request(request_id);
                    }
                }
            }
            _ => {
                debug!(target: LOG_TARGET, "file download event {:?}",  event);
            }
        }
    }

    /// 处理元数据下载事件
    pub(super) async fn process_metadata_download_event(
        &mut self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        event: request_response::Event<MetadataDownloadRequest, MetadataDownloadResponse>,
    ) {
        match event {
            request_response::Event::Message {
                peer,
                connection_id,
                message,
            } => {
                info!(target: LOG_TARGET, "Received metadata download request from peer: {:?}, connection_id: {:?}, message: {:?}", peer, connection_id, message);
                match message {
                    request_response::Message::Request {
                        request_id,
                        request,
                        channel,
                    } => {
                        info!(target: LOG_TARGET, "Received metadata download request from peer: {:?}, connection_id: {:?}, request_id: {:?}, request: {:?}", peer, connection_id, request_id, request);
                        // check if we have the requested file fully available
                        match self.file_store.published_file_exists(request.file_id) {
                            Ok(true) => {
                                info!(target: LOG_TARGET, "File {:?} exists", request.file_id);
                                match self
                                    .file_store
                                    .published_file_metadata_path(request.file_id)
                                {
                                    Ok(path) => {
                                        info!(target: LOG_TARGET, "File {:?} metadata path: {:?}", request.file_id, path);
                                        match fs::read(path).await {
                                            Ok(metadata) => {
                                                info!(target: LOG_TARGET, "File {:?} metadata: {:?}", request.file_id, metadata);
                                                if let Err(e) = swarm
                                                    .behaviour_mut()
                                                    .metadata_download
                                                    .send_response(
                                                        channel,
                                                        MetadataDownloadResponse::Success(metadata),
                                                    )
                                                {
                                                    error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                                }
                                            }
                                            Err(e) => {
                                                if let Err(e) = swarm
                                                    .behaviour_mut()
                                                    .metadata_download
                                                    .send_response(
                                                        channel,
                                                        MetadataDownloadResponse::Error(format!(
                                                            "Failed to read file metadata: {:?}",
                                                            e
                                                        )),
                                                    )
                                                {
                                                    error!(target: LOG_TARGET, "Failed to send metadata download response: {:?}", e);
                                                }
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        if let Err(e) =
                                            swarm.behaviour_mut().metadata_download.send_response(
                                                channel,
                                                MetadataDownloadResponse::Error(format!(
                                                    "Failed to get file metadata path: {:?}",
                                                    e
                                                )),
                                            )
                                        {
                                            error!(target: LOG_TARGET, "Failed to check file existence: {:?}", e);
                                        }
                                    }
                                }
                            }
                            Ok(false) => {
                                info!(target: LOG_TARGET, "File {:?} does not exist", request.file_id);
                            }
                            Err(e) => {
                                if let Err(e) =
                                    swarm.behaviour_mut().metadata_download.send_response(
                                        channel,
                                        MetadataDownloadResponse::Error(format!(
                                            "File {:?} does not exist: {:?}",
                                            request.file_id, e
                                        )),
                                    )
                                {
                                    error!(target: LOG_TARGET, "Failed to check file existence: {:?}", e);
                                }
                            }
                        }
                    }
                    request_response::Message::Response {
                        request_id,
                        response,
                    } => {
                        info!(target: LOG_TARGET, "Received metadata download response from peer: {:?}, connection_id: {:?}, request_id: {:?}, response: {:?}", peer, connection_id, request_id, response);

                        match response {
                            MetadataDownloadResponse::Success(raw_metadata) => {
                                match FileProcessResult::try_from(raw_metadata.to_vec()) {
                                    Ok(metadata) => {
                                        info!(target: LOG_TARGET, "New metadata download: request_id {:?} filename {:?}, merkle root ({:?})", request_id, metadata.original_file_name, metadata.merkle_root);

                                        if let Some(metadata_request_data) = self
                                            .find_metadata_download_request_with_request_id_mut(
                                                request_id,
                                            )
                                        {
                                            info!(target: LOG_TARGET, "Send metadata download response, request_id: {:?}, metadata download success, sender: {:?}", request_id, metadata_request_data.result);
                                            // 这里使用mem::take取result的值，如果result有值，则该值被取出，result赋值为None
                                            // 或者使用 metadata_request_data.take() 也可以，底层是调用了 mem::replace()
                                            if let Some(sender) =
                                                mem::take(&mut metadata_request_data.result)
                                            {
                                                if let Err(e) = sender.send(metadata) {
                                                    info!(target: LOG_TARGET, "Failed to send metadata to request result channel: {:?}", e);
                                                    return;
                                                }
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        error!(target: LOG_TARGET, "Failed to parse metadata from bytes: {:?}", e);
                                    }
                                }
                            }
                            MetadataDownloadResponse::Error(err) => {
                                debug!(target: LOG_TARGET, "Kademlia outbound query progressed, request_id: {:?}, metadata download error: {:?}", request_id, err);
                            }
                        }

                        self.remove_metadata_download_request(request_id);
                    }
                }
            }
            _ => {
                debug!(target: LOG_TARGET, "metadata download event {:?}",  event);
            }
        }
    }

    /// 处理 kademlia 事件
    pub(super) fn process_kademlia_event(
        &mut self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        event: kad::Event,
    ) {
        match event {
            kad::Event::OutboundQueryProgressed {
                id: query_id,
                result,
                ..
            } => {
                info!(target: LOG_TARGET, "Kademlia outbound query progressed, query_id: {:?}, result: {:?}", query_id, result);
                if let QueryResult::GetProviders(Ok(providers)) = result {
                    self.handle_query_progressed(swarm, query_id, providers);
                };
            }
            _ => {
                debug!(target: LOG_TARGET, "kademlia event {:?}",  event);
            }
        }
    }

    fn handle_query_progressed(
        &mut self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        query_id: QueryId,
        providers: GetProvidersOk,
    ) {
        self.handle_metadata_download(swarm, query_id, providers.clone());
        self.handle_file_chunk_download(swarm, query_id, providers);
    }

    fn handle_metadata_download(
        &mut self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        query_id: QueryId,
        providers: GetProvidersOk,
    ) {
        let metadata_request_data = self.find_metadata_download_request_with_query_id(query_id);
        if metadata_request_data.is_none() {
            info!(target: LOG_TARGET, "Kademlia outbound query progressed, id: {:?}, cannot find metadata request data", query_id);
            return;
        }
        let metadata_request_data = metadata_request_data.unwrap();
        if metadata_request_data.download_metadata_request_id.is_some() {
            // 请求已经发送了，不需要再发送
            info!(target: LOG_TARGET, "Kademlia outbound query progressed, id: {:?}, request already sent", query_id);
            return;
        }

        let peer_id = match providers {
            kad::GetProvidersOk::FoundProviders {
                key: _key,
                providers,
            } => {
                // providers 是一个列表，但我们只取第一个(也就是最近的一个)
                // 数据发送后，这个节点会在 metadata download 事件中收到数据，并进行下载metadata的操作
                providers.iter().next().cloned()
            }
            kad::GetProvidersOk::FinishedWithNoAdditionalRecord { closest_peers } => {
                if closest_peers.is_empty() {
                    None
                } else {
                    // closest_peers 是一个列表，但我们只取第一个(也就是最近的一个)
                    // 数据发送后，这个节点会在 metadata download 事件中收到数据，并进行下载metadata的操作
                    closest_peers.first().cloned()
                }
            }
        };

        info!(target: LOG_TARGET, "Kademlia outbound query found peer id: {:?}", peer_id);

        if let Some(peer_id) = peer_id {
            info!(target: LOG_TARGET, "Kademlia outbound query found peer id: {:?}, send metadata download request: {:?}", peer_id, metadata_request_data.request);
            let request_id = swarm
                .behaviour_mut()
                .metadata_download
                .send_request(&peer_id, metadata_request_data.request.clone());
            info!(target: LOG_TARGET, "Kademlia outbound query found peer id: {:?}, send request request_id: {:?}", peer_id, request_id);
            self.set_metadata_download_request_request_id(query_id, request_id);
        }
    }

    fn handle_file_chunk_download(
        &mut self,
        swarm: &mut Swarm<P2pNetworkBehavior>,
        query_id: QueryId,
        providers: GetProvidersOk,
    ) {
        let file_chunk_request_data = self.find_file_chunk_download_request_with_query_id(query_id);
        if file_chunk_request_data.is_none() {
            info!(target: LOG_TARGET, "Kademlia outbound query progressed, id: {:?}, cannot find file chunk request data", query_id);
            return;
        }
        let file_chunk_request_data = file_chunk_request_data.unwrap();
        if file_chunk_request_data
            .download_file_chunk_request_id
            .is_some()
        {
            // 请求已经发送了，不需要再发送
            info!(target: LOG_TARGET, "Kademlia outbound query progressed, id: {:?}, request already sent", query_id);
            return;
        }

        let peer_id = match providers {
            kad::GetProvidersOk::FoundProviders {
                key: _key,
                providers,
            } => {
                // providers 是一个列表，但我们只取第一个(也就是最近的一个)
                // 数据发送后，这个节点会在 file chunk download 事件中收到数据，并进行下载file chunk的操作
                providers.iter().next().cloned()
            }
            kad::GetProvidersOk::FinishedWithNoAdditionalRecord { closest_peers } => {
                if closest_peers.is_empty() {
                    None
                } else {
                    // closest_peers 是一个列表，但我们只取第一个(也就是最近的一个)
                    // 数据发送后，这个节点会在 file chunk download 事件中收到数据，并进行下载file chunk的操作
                    closest_peers.first().cloned()
                }
            }
        };

        info!(target: LOG_TARGET, "Kademlia outbound query found peer id: {:?}", peer_id);

        if let Some(peer_id) = peer_id {
            info!(target: LOG_TARGET, "Kademlia outbound query found peer id: {:?}, send file chunk download request: {:?}", peer_id, file_chunk_request_data.request);
            let request_id = swarm
                .behaviour_mut()
                .file_download
                .send_request(&peer_id, file_chunk_request_data.request.clone());
            info!(target: LOG_TARGET, "Kademlia outbound query found peer id: {:?}, send request request_id: {:?}", peer_id, request_id);
            self.set_file_chunk_download_request_request_id(query_id, request_id);
        }
    }
}
